package com.netty.demo.binary.factorial;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.math.BigInteger;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author shengtao
 * @Description:
 * @date 2018/09/10 13:58
 */
public class FactorialClientHandler extends ChannelHandlerAdapter {

    private ChannelHandlerContext ctx;

    private int recievedMessages;

    private int next=1;

    private final BlockingQueue<BigInteger> auswer = new LinkedBlockingQueue<>();

    public BigInteger getFactorial(){
        boolean interrupted = false;
        try {
            for (;;){
                try {
                    return auswer.take();
                }catch (InterruptedException e){
                    interrupted = true;
                    e.printStackTrace();
                }
            }
        }finally {
            if(interrupted){
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
        sendNumbers();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        BigInteger message = (BigInteger)msg;
//        log.error("Send mail failed, error message is : {} \n", e.getMessage());
        System.out.println("====="+msg);
        recievedMessages ++;
        if (recievedMessages == FactorialClient.COUNT) {
            ctx.channel().close().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    boolean offered = auswer.offer(message);
                    assert offered;
                }
            });
        }
    }

    private void sendNumbers() {
        ChannelFuture future = null;
        for (int i = 1;i<4096 && next < FactorialClient.COUNT;i++){
            future = ctx.write(Integer.valueOf(i));
            next++;
        }
        if(next <= FactorialClient.COUNT){
            assert future != null;
            future.addListener(listener);
        }
        ctx.flush();
    }

    private final ChannelFutureListener listener = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            if(channelFuture.isSuccess()){
                sendNumbers();
            }else {
                channelFuture.cause().printStackTrace();
                channelFuture.channel().close();
            }
        }

    };

}
